package es

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"reflect"
	"strconv"
	"strings"
	"time"

	//"github.com/elastic/go-elasticsearch/esapi"
	//"github.com/elastic/go-elasticsearch/v7"
	"github.com/elastic/go-elasticsearch/v8"
	"github.com/elastic/go-elasticsearch/v8/esapi"
	"github.com/toolkits/pkg/logger"
)

var ES *elasticsearch.Client
var ENV_INDEX string = "environment_sla"
var ENV_INDEX_HISTORY string = "environment-*"
var EVENT_INDEX string = "environment_event"

type EsSection struct {
	Enabled      bool   `yaml:"enabled"`
	Address      string `yaml:"address"`
	Username     string `yaml:"username"`
	Password     string `yaml:"password"`
	EnvIndex     string `yaml:"envindex"`
	EnvPattern   string `yaml:"envpattern"`
	EventIndex   string `yaml:"eventindex"`
	EventPattern string `yaml:"eventpattern"`
	MaxConn      int    `yaml:"maxconn"`
	Timeout      int    `yaml:"timeout"`
}

func NewConn(cfg EsSection) error {
	escfg := elasticsearch.Config{
		Addresses: []string{
			cfg.Address,
		},
		Username: cfg.Username,
		Password: cfg.Password,
		//EnableRetryOnTimeout:  true,
		MaxRetries:            5,
		DiscoverNodesOnStart:  true,
		DiscoverNodesInterval: 60 * time.Second,
		Transport: &http.Transport{
			ResponseHeaderTimeout: time.Duration(cfg.Timeout) * time.Second,
			DialContext:           (&net.Dialer{Timeout: time.Second}).DialContext,
		},
	}
	var err error
	ES, err = elasticsearch.NewClient(escfg)
	if err != nil {
		logger.Errorf("connect failed:%v", err)
		return err
	}
	res, err := ES.Info()
	if err != nil {
		logger.Errorf("es info get failed:%v", err)
		return err
	}
	logger.Infof("connect es resq: %+v", res)

	ENV_INDEX = cfg.EnvIndex
	ENV_INDEX_HISTORY = cfg.EnvPattern
	EVENT_INDEX = cfg.EventIndex

	logger.Infof("ENV_INDEX:%s ENV_INDEX_HISTORY:%s EVENT_INDEX:%s", ENV_INDEX, ENV_INDEX_HISTORY, EVENT_INDEX)

	return nil
}

func PutIndex(index string, doc string) error {
	/*
			jsonString, err := json.Marshal([]byte(doc))
		    if err != nil {
		        fmt.Println("json marshal failed", err)
		        return err
		    }
	*/
	request := esapi.IndexRequest{Index: index, Body: strings.NewReader(doc)}
	resp, err := request.Do(context.Background(), ES)
	if err != nil {
		logger.Errorf("PutIndex %s write es failed:%v, %+v", index, err, resp)
		return err
	}
	return nil
}

func UpdatebyHost(hostname string, index string, doc interface{}) error {
	body := map[string]interface{}{
		"doc_as_upsert": true,
		"doc":           doc,
	}
	jsonString, err := json.Marshal(body)
	if err != nil {
		logger.Errorf("json marshal failed:%+v", err)
		return err
	}

	request := esapi.UpdateRequest{Index: index, DocumentID: hostname, Body: strings.NewReader(string(jsonString))}
	resp, err := request.Do(context.Background(), ES)
	if err != nil {
		logger.Errorf("write es failed:%+v", resp)
		return err
	}
	defer resp.Body.Close()

	if resp.IsError() {
		var e map[string]interface{}
		if err := json.NewDecoder(resp.Body).Decode(&e); err != nil {
			logger.Errorf("[ES UPDATE] Error parsing the response error: %v body:%s", err, jsonString)
			return err
		}
		logger.Errorf("[ES UPDATE] [%s] error:%+v, index:%s id:%s, body:%s", resp.Status(), e["error"], index, hostname, jsonString)
		return fmt.Errorf("%#v", resp.Status())

	}
	return nil
}

func GetEnv(hostname string, index string) (e map[string]interface{}, err error) {
	request := esapi.GetRequest{Index: index, DocumentID: hostname}
	res, err := request.Do(context.Background(), ES)
	if err != nil {
		logger.Errorf("[ES Get] Error getting response: %v", err)
		return
	}
	defer res.Body.Close()

	e = map[string]interface{}{}
	if err = json.NewDecoder(res.Body).Decode(&e); err != nil {
		logger.Errorf("[ES Get] Error parsing the response body: %v", err)
		return
	}

	if res.IsError() {
		//logger.Errorf("[ES Get] [%s] %+v", res.Status(), e["error"])
		return e, fmt.Errorf("%#v", res.Status())
	}
	return
}

func Match(index string, matchType string, query map[string]interface{}, sourceName []string) ([]interface{}, error) {
	querys := map[string]interface{}{
		"_source": sourceName,
		"query": map[string]interface{}{
			matchType: query,
		},
	}
	source := []interface{}{}

	//开始查询
	var buf bytes.Buffer
	if err := json.NewEncoder(&buf).Encode(querys); err != nil {
		logger.Errorf("[ES Match] Error encoding query: %v", err)
		return source, err
	}

	// 执行搜索请求.
	res, err := ES.Search(
		ES.Search.WithContext(context.Background()),
		ES.Search.WithIndex(index),
		ES.Search.WithBody(&buf),
		ES.Search.WithTrackTotalHits(true),
		ES.Search.WithPretty(),
	)
	if err != nil {
		logger.Errorf("[ES Match] Error getting response: %v", err)
		return source, err
	}
	defer res.Body.Close()

	if res.IsError() {
		var e map[string]interface{}
		if err := json.NewDecoder(res.Body).Decode(&e); err != nil {
			logger.Errorf("[ES Match] Error parsing the response error body: %v", err)
			return source, err
		}
		//logger.Errorf("[ES Get] [%s] %+v", res.Status(), e["error"])
		return source, fmt.Errorf("%v", res.Status())
	}

	var r map[string]interface{}
	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
		logger.Errorf("[ES Match] Error parsing the response body: %v", err)
		return source, err
	}

	/* 打印响应状态，结果数和请求持续时间.
	   fmt.Printf(
	       "[%s] %d hits; took: %dms\n",
	       res.Status(),
	       int(r["hits"].(map[string]interface{})["total"].(map[string]interface{})["value"].(float64)),
	       int(r["took"].(float64)),
	   )
	*/

	// 打印每次匹配的ID和文档来源.
	if s, ok := r["hits"].(map[string]interface{})["hits"]; ok {
		if reflect.ValueOf(s).Kind() == reflect.Slice {
			for _, hit := range r["hits"].(map[string]interface{})["hits"].([]interface{}) {
				source = append(source, hit.(map[string]interface{})["_source"])
			}
		}
	}

	return source, nil
}

func Query(index string, query map[string]interface{}) (r map[string]interface{}, err error) {
	r = map[string]interface{}{}
	// 建立请求主体.
	var buf bytes.Buffer
	if err = json.NewEncoder(&buf).Encode(query); err != nil {
		logger.Errorf("[ES Query] Error encoding query: %v", err)
		return r, err
	}

	// 执行搜索请求.
	res, err := ES.Search(
		ES.Search.WithContext(context.Background()),
		ES.Search.WithIndex(index),
		ES.Search.WithBody(&buf),
		ES.Search.WithTrackTotalHits(true),
		ES.Search.WithPretty(),
	)
	if err != nil {
		logger.Errorf("[ES Query] Error getting response: %v", err)
		return r, err
	}
	defer res.Body.Close()

	/*
	   buf2 := new(bytes.Buffer)
	   buf2.ReadFrom(res.Body)
	   newStr := buf2.String()
	   fmt.Println("Query response:", newStr)
	*/

	if res.IsError() {
		var e map[string]interface{}
		if err = json.NewDecoder(res.Body).Decode(&e); err != nil {
			logger.Errorf("[ES Query] Error parsing the error body: %v", err)
			return r, err
		}
		//logger.Errorf("[ES Get] [%v] %+v", res.Status(), e["error"])
		return r, fmt.Errorf("%#v", res.Status())
	}

	if err = json.NewDecoder(res.Body).Decode(&r); err != nil {
		logger.Errorf("[ES Query] Error parsing the response body: %v", err)
		return r, err
	}
	logger.Debugf("[ES Query] response from query: %+v", r)

	return r, nil

}

func Consistency(node string) (sla map[string]float64, err error) {
	query := `{
  "size": 0,
  "query": {
    "prefix": {
      "node.keyword": {
        "value": "%s"
      }
    }
  },
 "aggs": {
    "generalconsistency": {
      "terms": {
        "field": "generalstatus",
        "size": 10
      }
    },
    "serviceconsistency": {
      "terms": {
        "field": "servicestatus",
        "size": 10
      }
    },
    "personalconsistency": {
      "terms": {
        "field": "personalstatus",
        "size": 10
      }
    }
  }
}`
	query = fmt.Sprintf(query, node)

	request := esapi.SearchRequest{Index: []string{ENV_INDEX}, Body: strings.NewReader(query)}
	res, err := request.Do(context.Background(), ES)
	if err != nil {
		logger.Errorf("write date query failed: %v, res:%+v", err, res)
		return
	}

	if res.IsError() {
		var e map[string]interface{}
		if err = json.NewDecoder(res.Body).Decode(&e); err != nil {
			logger.Errorf("[ES DateQuery] Error parsing the response error body: %v", err)
			return
		}
		//logger.Errorf("[ES Get] [%s] %+v\n", res.Status(), e["error"])
		err = fmt.Errorf("%#v", res.Status())
		return
	}

	var r map[string]interface{}
	if err = json.NewDecoder(res.Body).Decode(&r); err != nil {
		logger.Errorf("[ES DateQuery] Error parsing the response body: %v", err)
		return
	}

	sla = map[string]float64{}

	// 打印每次匹配的ID和文档来源.
	if _, ok := r["hits"]; ok {
		t1 := r["hits"].(map[string]interface{})["total"].(map[string]interface{})["value"].(float64)
		if t1 == 0 {
			logger.Errorf("[ES Consistency] get nothing: %#v", r)
			return sla, fmt.Errorf("total is zero")
		}
		sla["total"] = t1
	} else {
		//logger.Errorf("[ES DateQuery] cannot find data: %v", r)
		return sla, fmt.Errorf("total is zero")
	}

	var maxCount float64 = -1
	var total float64
	var docCount float64
	var generalstatus float64
	var personalstatus float64
	var envconsistency float64
	var succ float64
	var generalSuccrate float64
	var syscheck bool
	var personalcheck bool
	var sercivecheck bool
	var sysfail float64
	var servicefail float64

	for _, hit := range r["aggregations"].(map[string]interface{})["generalconsistency"].(map[string]interface{})["buckets"].([]interface{}) {
		docCount = hit.(map[string]interface{})["doc_count"].(float64)
		checkstatus := hit.(map[string]interface{})["key"].(float64)
		if checkstatus == 1 {
			succ = docCount
			syscheck = true
		} else if checkstatus == 2 {
			sysfail = docCount
			syscheck = true
		}
		if docCount > maxCount {
			maxCount = docCount
		}
		total = total + docCount
	}
	if total != 0 {
		generalstatus, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", maxCount/total), 64)
		generalSuccrate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", succ/total), 64)
	} else {
		generalstatus = 0
		generalSuccrate = 0
	}

	total = 0
	succ = 0
	maxCount = -1
	var personalSuccrate float64

	for _, hit := range r["aggregations"].(map[string]interface{})["personalconsistency"].(map[string]interface{})["buckets"].([]interface{}) {
		docCount = hit.(map[string]interface{})["doc_count"].(float64)
		checkstatus := hit.(map[string]interface{})["key"].(float64)
		if checkstatus == 1 {
			succ = docCount
		}
		if checkstatus != 0 {
			personalcheck = true
		}
		if docCount > maxCount {
			maxCount = docCount
		}
		total = total + docCount
	}
	if total != 0 {
		personalstatus, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", maxCount/total), 64)
		personalSuccrate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", succ/total), 64)
	} else {
		personalstatus = 0
		personalSuccrate = 0
	}

	if syscheck {
		if personalcheck {
			envconsistency, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", (personalstatus+generalstatus)/2), 64)
			generalSuccrate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", (personalSuccrate+generalSuccrate)/2), 64)
		} else {
			envconsistency = generalstatus
			//generalSuccrate = personalSuccrate
		}
	} else {
		if personalcheck {
			envconsistency = personalstatus
			generalSuccrate = personalSuccrate
		} else {
			envconsistency = 0
		}
	}

	sla["environment_consistency"] = envconsistency
	sla["environment_succrate"] = generalSuccrate
	sla["environment_failnum"] = sysfail

	total = 0
	succ = 0
	maxCount = -1

	for _, hit := range r["aggregations"].(map[string]interface{})["serviceconsistency"].(map[string]interface{})["buckets"].([]interface{}) {
		docCount = hit.(map[string]interface{})["doc_count"].(float64)
		checkstatus := hit.(map[string]interface{})["key"].(float64)
		if checkstatus == 1 {
			succ = docCount
			sercivecheck = true
		} else if checkstatus == 2 {
			servicefail = docCount
			sercivecheck = true
		}
		if docCount > maxCount {
			maxCount = docCount
		}
		total = total + docCount
	}
	if total != 0 {
		sla["service_consistency"], _ = strconv.ParseFloat(fmt.Sprintf("%.2f", maxCount/total), 64)
		sla["service_succrate"], _ = strconv.ParseFloat(fmt.Sprintf("%.2f", succ/total), 64)
	} else {
		sla["service_consistency"] = 0
		sla["service_succrate"] = 0
	}
	sla["service_failnum"] = servicefail

	var serviceSla float64
	if sercivecheck {
		if syscheck || personalcheck {
			serviceSla = (sla["service_succrate"] + generalSuccrate) / 2
			serviceSla, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", serviceSla*100), 64)
		} else {
			serviceSla = sla["service_succrate"] * 100
		}
	} else {
		if syscheck || personalcheck {
			serviceSla = generalSuccrate * 100
		}
	}

	sla["sla"] = serviceSla
	return
}

func DateConsistency(node string) (result map[string]interface{}, err error) {
	query := `{
  "size": 0, 
  "query": {
    "bool": {
      "must": [
        {
          "prefix": {
            "node.keyword": {
              "value": "%s"
            }
          }
        }
      ], 
      "filter": [
        {
          "range": {
            "update_time": {
              "gte": "now-7d/d",
              "lt":  "now"
    	      }
          }
        }
      ]
    }
  }, 
  "aggs": {
    "time_bucket": {
      "date_histogram": {
        "field": "update_time",
        "calendar_interval": "day"
      },
      "aggs": {
        "generalstatus": {
          "terms": {
            "field": "generalstatus",
            "size": 10
          }
        },
        "personalstatus": {
          "terms": {
            "field": "personalstatus",
            "size": 10
          }
        },
        "servicestatus": {
          "terms": {
            "field": "servicestatus",
            "size": 10
          }
        }
      }
    }
  }
}`

	query = fmt.Sprintf(query, node)

	request := esapi.SearchRequest{Index: []string{ENV_INDEX_HISTORY}, Body: strings.NewReader(query)}
	res, err := request.Do(context.Background(), ES)
	if err != nil {
		logger.Errorf("write date query failed:%v, res:%+v", err, res)
		return
	}
	if res.IsError() {
		var e map[string]interface{}
		if err = json.NewDecoder(res.Body).Decode(&e); err != nil {
			logger.Errorf("[ES DateQuery] Error parsing the response error body: %v", err)
			return
		}

		logger.Errorf("[ES Get] [%s] %+v; query is:%s index is:%s ", res.Status(), e["error"], query, ENV_INDEX_HISTORY)
		err = fmt.Errorf("check failed: %#v", res.Status())
		return
	}

	var r map[string]interface{}
	if err = json.NewDecoder(res.Body).Decode(&r); err != nil {
		logger.Errorf("[ES DateQuery] Error parsing the response body: %v", err)
		return
	}

	// 打印每次匹配的ID和文档来源.
	if _, ok := r["hits"]; ok {
		t1 := r["hits"].(map[string]interface{})["total"].(map[string]interface{})["value"].(float64)
		if t1 == 0 {
			logger.Errorf("[ES DateQuery] get nothing: %#v", r)
			return result, fmt.Errorf("get nothing")
		}
	} else {
		//logger.Errorf("[ES DateQuery] get nothing: %#v", r)
		return result, fmt.Errorf("get nothing")
	}

	serviceConsist := map[string]float64{}
	environmentConsist := map[string]float64{}
	serviceSla := map[string]float64{}
	environmentSuccrate := map[string]float64{}
	serviceSuccrate := map[string]float64{}

	var maxCount float64 = -1
	var total float64
	var personalConsist float64 = 0
	var generalConsist float64 = 0
	var personalSuccrate float64 = 0
	var econsist float64
	var sconsist float64
	var esuccrate float64
	var ssuccrate float64
	var date string
	var sla float64
	var succ float64
	var syscheck bool
	var personalcheck bool
	var sercivecheck bool

	for _, hit := range r["aggregations"].(map[string]interface{})["time_bucket"].(map[string]interface{})["buckets"].([]interface{}) {
		date = hit.(map[string]interface{})["key_as_string"].(string)
		syscheck = false
		personalcheck = false
		sercivecheck = false
		//计算历史自定义
		personalstatus := hit.(map[string]interface{})["personalstatus"].(map[string]interface{})["buckets"].([]interface{})
		if len(personalstatus) != 0 {
			for _, v := range personalstatus {
				docCount := v.(map[string]interface{})["doc_count"].(float64)
				checkstatus := v.(map[string]interface{})["key"].(float64)
				if checkstatus == 1 {
					succ = docCount
				}
				if checkstatus != 0 {
					personalcheck = true
				}
				if docCount > maxCount {
					maxCount = docCount
				}
				total = total + docCount
			}

			if total != 0 {
				personalConsist, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", maxCount/total), 64)
				personalSuccrate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", succ/total), 64)
			}
		}

		//计算历史通用一致性
		generalstatus := hit.(map[string]interface{})["generalstatus"].(map[string]interface{})["buckets"].([]interface{})
		if len(generalstatus) != 0 {
			maxCount = -1
			total = 0
			succ = 0
			for _, v := range generalstatus {
				docCount := v.(map[string]interface{})["doc_count"].(float64)
				checkstatus := v.(map[string]interface{})["key"].(float64)
				if checkstatus == 1 {
					succ = docCount
				}
				if checkstatus != 0 {
					syscheck = true
				}
				if docCount > maxCount {
					maxCount = docCount
				}
				total = total + docCount
			}

			if total != 0 {
				generalConsist, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", maxCount/total), 64)
				esuccrate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", succ/total), 64)
			}
		}

		//计算环境整体一致性
		if syscheck {
			if personalcheck {
				econsist = (personalConsist + generalConsist) / 2
				esuccrate = (esuccrate + personalSuccrate) / 2
			} else {
				econsist = generalConsist
			}
		} else {
			if personalcheck {
				econsist = personalConsist
				esuccrate = personalSuccrate
			} else {
				econsist = 0
				esuccrate = 0
			}
		}

		econsist, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", econsist), 64)
		esuccrate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", esuccrate), 64)
		environmentConsist[date] = econsist
		environmentSuccrate[date] = esuccrate

		servicestatus := hit.(map[string]interface{})["servicestatus"].(map[string]interface{})["buckets"].([]interface{})
		if len(servicestatus) != 0 {
			maxCount = -1
			total = 0
			succ = 0
			for _, v := range servicestatus {
				docCount := v.(map[string]interface{})["doc_count"].(float64)
				checkstatus := v.(map[string]interface{})["key"].(float64)
				if checkstatus == 1 {
					succ = docCount
				}
				if checkstatus != 0 {
					sercivecheck = true
				}

				if docCount > maxCount {
					maxCount = docCount
				}
				total = total + docCount
			}

			if total != 0 {
				ssuccrate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", succ/total), 64)
				sconsist, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", maxCount/total), 64)
			}
		}
		serviceConsist[date] = sconsist
		serviceSuccrate[date] = ssuccrate

		if sercivecheck {
			if syscheck || personalcheck {
				sla = (esuccrate + ssuccrate) * 100 / 2
			} else {
				sla = ssuccrate * 100
			}

		} else {
			if syscheck || personalcheck {
				sla = esuccrate * 100
			}
		}

		sla, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", sla), 64)
		serviceSla[date] = sla

	}

	result = map[string]interface{}{}
	result["environment_consistency"] = environmentConsist
	result["service_consistency"] = serviceConsist
	result["service_sla"] = serviceSla
	result["environment_succrate"] = environmentSuccrate
	result["service_succrate"] = serviceSuccrate
	return
}
